package com.lagou.rpc.consumer;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.common.Metric;
import com.lagou.rpc.zookeeper.NodeChangeListener;
import com.lagou.rpc.common.RequestMetrics;
import com.lagou.rpc.zookeeper.ZookeeperClient;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 消费者
 */
public class Consumer implements NodeChangeListener {
    private ZookeeperClient zookeeperClient;
    //服务端信息列表：存储格式<服务名称，RPC客户端列表>
    public static final Map<String, List<RpcClient>> RPC_CLIENT_POOL = new HashMap<>();

    /**
     * 初始化消费端
     *
     * @param zookeeperClient
     * @param instanceCacheMap
     */
    public Consumer(ZookeeperClient zookeeperClient, Map<String, Object> instanceCacheMap) {
        this.zookeeperClient = zookeeperClient;
        //遍历客户端所有需要调用的服务
        instanceCacheMap.entrySet().stream().forEach(entry -> {
            String serviceName = entry.getKey();
            //根据服务名拼接路径
            //通过zookeeperClient查询服务
            List<String> serviceList = zookeeperClient.discovery(serviceName);
            //建立连接，并将连接放到RPC_CLIENT_POOL
            if (serviceList != null && serviceList.size() > 0) {
                //某个服务的连接集合
                List<RpcClient> rpcClients = new ArrayList<>();
                for (String serviceInfo : serviceList) {
                    String[] split = serviceInfo.split(":");
                    //和服务端建立连接
                    RpcClient rpcClient = new RpcClient(split[0], Integer.parseInt(split[1]));
                    rpcClients.add(rpcClient);
                    //添加到客户端上报集合
                    RequestMetrics.getInstance().addNode(split[0], Integer.parseInt(split[1]));
                }
                RPC_CLIENT_POOL.put(serviceName, rpcClients);
                System.out.println("RPC_CLIENT_POOL中的信息：" + RPC_CLIENT_POOL);
            }
        });
        //将当前消费者加入到监听器列表中
        zookeeperClient.addListener(this);
    }

    /**
     * 创建并返回代理对象
     *
     * @param serviceClass
     * @return
     */
    public Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass}, new InvocationHandler() {
            @Override
            public Object invoke(Object o, Method method, Object[] args) throws Throwable {
                //1.封装request对象
                RpcRequest rpcRequest = new RpcRequest();
                rpcRequest.setRequestId(UUID.randomUUID().toString());
                rpcRequest.setClassName(method.getDeclaringClass().getName());
                rpcRequest.setParameterTypes(method.getParameterTypes());
                rpcRequest.setParameters(args);
                rpcRequest.setMethodName(method.getName());
                //2.创建RpcClient对象
//                RpcClient rpcClient = new RpcClient("127.0.0.1", 8899);

                //负载均衡
                RpcClient rpcClient = route(method.getDeclaringClass().getName());
                //判断rpcClient是否为空，以及是否为有效链接
                if (null == rpcClient) {
                    System.out.println("客户端为空，重新获取客户端");
                    Thread.sleep(3000);
                    rpcClient = route(method.getDeclaringClass().getName());
                }
                try {
                    //3.发送消息
                    //Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                    RpcResponse rpcResponse = (RpcResponse) rpcClient.send(rpcRequest);
                    //RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                    if (rpcResponse.getError() != null) {
                        throw new RuntimeException(rpcResponse.getError());
                    }
                    Object result = rpcResponse.getResult();
                    //4.返回结果
                    return JSON.parseObject(result.toString(), method.getReturnType());
                } catch (Exception e) {
                    if (e.getClass().getName().equals("java.nio.channels.ClosedChannelException")) {
                        System.out.println("调用服务发生异常:" + e.getMessage());
                        e.printStackTrace();
                        Thread.sleep(3000);
                        RpcClient otherRpcClient = route(method.getDeclaringClass().getName());
                        if (null == otherRpcClient) {
                            return null;
                        }
                        RpcResponse rpcResponse = (RpcResponse) rpcClient.send(rpcRequest);
                        //4.返回结果
                        return JSON.parseObject(rpcResponse.getResult().toString(), method.getReturnType());
                    }
                    throw e;
                }
            }
        });
    }

    /**
     * 获取路由客户端
     *
     * @return
     */
    private RpcClient route(String serviceName) {
        //根据服务查找本地缓存的客户端列表
        List<RpcClient> rpcClients = RPC_CLIENT_POOL.get(serviceName);
        System.out.println("可供" + serviceName + "调用的服务列表：");
        //信息打印
        for (int i = 0; i < rpcClients.size(); i++) {
            RpcClient rpcClient = rpcClients.get(i);
            System.out.println("        " +rpcClient.getIp() + ":" + rpcClient.getPort() + ";");
        }
        if (rpcClients == null) {
            return null;
        }
        //查找上报的所有服务端列表
        List<Metric> allMetrics = RequestMetrics.getInstance().getAllMetrics();
        //如果上报的列表为空则随机返回一个客户端
        if (allMetrics.isEmpty()) {
            return rpcClients.get(new Random().nextInt(rpcClients.size()));
        }
        //对上报的列表进行排序：排序后如果只有一个则查找对应的客户端返回，如果不止一个取前两个判断时间是否一致，不一致则取第一个，一致的话随机取一个
        Collections.sort(allMetrics);
        Metric metric0 = allMetrics.get(0);
        //判断是否只有一个
        if (allMetrics.size() == 1) {
            Stream<RpcClient> rpcClientStream = rpcClients.stream().filter(new Predicate<RpcClient>() {
                @Override
                public boolean test(RpcClient rpcClient) {
                    if (rpcClient.getIp().equals(metric0.getIp()) && rpcClient.getPort() == metric0.getPort()) {
                        return true;
                    }
                    return false;
                }
            });
            RpcClient rpcClient = rpcClientStream.collect(Collectors.toList()).get(0);
            return rpcClient;
        }

        Metric metric1 = allMetrics.get(1);
        //如果两个耗时相同则随机返回；
        if (metric0.getCost().equals(metric1.getCost())) {
            return rpcClients.get(new Random().nextInt(rpcClients.size()));
        }
        //耗时不同则返回第一个耗时最短的客户端
        Stream<RpcClient> rpcClientStream = rpcClients.stream().filter(new Predicate<RpcClient>() {
            @Override
            public boolean test(RpcClient rpcClient) {
                if (rpcClient.getIp().equals(metric0.getIp()) && rpcClient.getPort() == metric0.getPort()) {
                    return true;
                }
                return false;
            }
        });
        RpcClient rpcClient = rpcClientStream.collect(Collectors.toList()).get(0);
        return rpcClient;
    }

    /**
     * 事件通知处理
     * @param service                服务名称
     * @param pathChildrenCacheEvent 通知事件
     */
    @Override
    public void notify(String service, PathChildrenCacheEvent pathChildrenCacheEvent) {
        //获取变更的节点路径
        String path = pathChildrenCacheEvent.getData().getPath();
        //得到ip和端口
        String serviceAddr = path.substring(path.lastIndexOf("/") + 1);
        String[] addr = serviceAddr.split(":");
        String ip = addr[0];
        int port = Integer.parseInt(addr[1]);

        //变更的节点类型
        PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
        List<RpcClient> rpcClients = RPC_CLIENT_POOL.get(service);

        //服务端节点增加：新增rpc连接
        if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(type)) {
            //判断节点列表是否为空
            if (rpcClients == null) {
                rpcClients = new ArrayList<>();
            }
            //与新增的节点建立连接
            RpcClient rpcClient = new RpcClient(ip, port);
            //添加到调用列表中
            rpcClients.add(rpcClient);
            RPC_CLIENT_POOL.put(service, rpcClients);
            //服务端节点增加时添加节点耗时记录
            RequestMetrics.getInstance().addNode(ip, port);
            System.out.println("新增节点： " + serviceAddr);
        } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(type)) {
            //服务端节点移除：断开rpc连接
            if (rpcClients == null) {
                return;
            }
            for (RpcClient rpcClient : rpcClients) {
                if (rpcClient.getIp().equals(ip) && rpcClient.getPort() == port) {
                    //列表中移除
                    rpcClients.remove(rpcClient);
                    System.out.println("移除节点： " + serviceAddr);
                    //服务端移除时移除节点耗时记录
                    RequestMetrics.getInstance().removeNode(ip, port);
                }
            }
        }
        //节点信息变更后
        System.out.println("RPC_CLIENT_POOL中节点信息：" + RPC_CLIENT_POOL);


    }
}
